package com.dmp.storm.bolt.dynamicdb;

import com.dmp.common.date.DateUtils;
import com.dmp.storm.bolt.BaseMapBolt;
import com.dmp.util.StrUtils;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;


/**
 * @author guozh3
 * 根据用户id将话单缓存到集合中
 */
public abstract class DyMapCacheBolt extends BaseMapBolt{
	private static final long serialVersionUID = 1L;
	private List<Map<String, Object>> userRecodeList;
	private ScheduledExecutorService scheduExec;
	private int timeOut = 60;
	private int recordSize = 500;
	private volatile boolean flag = false;
	private String nowDay = "";
	
	
	@SuppressWarnings("rawtypes")
	public ScheduledFuture future;
	private int i = 0;
	@SuppressWarnings("rawtypes")
	private ScheduledFuture startTask(){
		return scheduExec.scheduleWithFixedDelay(new Runnable() {
			@Override
			public void run() {
				flag = true;
				core();
			}
		}, timeOut*1000, timeOut*1000 ,TimeUnit.MILLISECONDS);
	}
	@Override
	public void prepare() {
		userRecodeList = new ArrayList<Map<String, Object>>();
		scheduExec = Executors.newSingleThreadScheduledExecutor();
		future = startTask();
		
		setNowDay(DateUtils.formatDate(new Date(System.currentTimeMillis()), "yyyyMMdd"));
	}
	@Override
	public Map<String, Object> process(Map<String, Object> map) {
		String d = DateUtils.formatDate(new Date(System.currentTimeMillis()), "yyyyMMdd");
		if(StrUtils.equals(d, this.nowDay)){
			i++;
			userRecodeList.add(map);
			if(i>=recordSize){
				this.flag = true;
				System.out.println("recodeSize = " + i);
			}
			if(flag){
				core();
			}
		}else{
			this.flag = true;
			this.core();
			i++;
			setNowDay(DateUtils.formatDate(new Date(System.currentTimeMillis()), "yyyyMMdd"));
			userRecodeList.add(map);
		}
		
		return map;
	}
	/**
	 * 核心处理方法
	 */
	private void core() {
		
		this.i=0;
		this.flag = false;
		if (!CollectionUtils.isEmpty(userRecodeList)) {
			processBatch(userRecodeList);
			userRecodeList = new ArrayList<Map<String, Object>>();
			future.cancel(true);
			future = startTask();
		}
	}
	public void setTimeOut(int timeOut) {
		this.timeOut = timeOut;
	}
	public void setRecordSize(int recordSize) {
		this.recordSize = recordSize;
	}
	public abstract void processBatch(List<Map<String, Object>> list);
	public String getNowDay() {
		return nowDay;
	}
	public void setNowDay(String nowDay) {
		this.nowDay = nowDay;
	}
	
}
